package trianing.vehicle;

import org.apache.commons.io.FileUtils;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.io.File;
import java.io.IOException;
import java.nio.charset.StandardCharsets;

/**
 * @desc
 * @Author xuyouchang
 * @Date 2022/2/17
 */
public class TrafficStat {
    public static void main(String[] args) throws IOException {
        String kafkaSql = FileUtils.readFileToString(new File(args[0]), StandardCharsets.UTF_8);
        String esSql = FileUtils.readFileToString(new File(args[1]), StandardCharsets.UTF_8);

        StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.createLocalEnvironment();
        StreamTableEnvironment streamTableEnvironment = StreamTableEnvironment.create(streamExecutionEnvironment);

        streamTableEnvironment.executeSql(kafkaSql);
        streamTableEnvironment.executeSql(esSql);

        String statSql = "select window_start, window_end,vehicle_type, plate_type, count(1) as stat from TABLE(TUMBLE(TABLE kafka_table, DESCRIPTOR(pass_time), INTERVAL '1' MINUTES)) " +
                "group by window_start, window_end, CUBE (vehicle_type, plate_type)";
        Table table = streamTableEnvironment.sqlQuery(statSql);
        table.executeInsert("es_table");
    }
}
